{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5f145c81-19c0-4e3f-865e-33d9227d076d",
   "metadata": {},
   "source": [
    "# Env"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "42f1098c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set base path\n",
    "BASE_PATH = \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "603d7c71-223d-4411-8e52-3ee5454d7b94",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "sys.path.append(os.path.abspath(\"..\")) # TreeOfLife-dev/notebooks\n",
    "# OSC's interactive session's PySpark Kernel set the current working directory to notebook directory\n",
    "\n",
    "from helpers.data_analysis import init_spark, create_freq, view_freq, check_sparsity\n",
    "from helpers.image import show_image_table, show_images, decode_image, find_image, show_image_interact\n",
    "from helpers.variables import COLS_TAXONOMIC\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql import types as T\n",
    "from pyspark.sql.functions import col, countDistinct, count, desc, broadcast, lower"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "7822d1fe-f0d0-410a-bc88-0cb085cf13ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Check if there is an active Spark session\n",
    "spark= SparkSession.getActiveSession()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b25afcba-c7d3-4e9b-843a-20ff4433fd16",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Columns of Interest\n",
    "spark_df = spark.read.parquet(f\"{BASE_PATH}/gbif/attributes/cols_of_interest\")\n",
    "\n",
    "# TOL-200M Metadata\n",
    "#spark_df_tol = spark.read.parquet(f\"{BASE_PATH}/TreeOfLife/metadata\")\n",
    "\n",
    "# Lookup Tables\n",
    "# Contain uuid and image file path\n",
    "# lookup_tbl = spark.read.parquet(f\"{BASE_PATH}/TreeOfLife/lookup_tables\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "63c6bc2a-ad4e-405b-8885-8a5997e39a0b",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "occ_df = spark.read.parquet(f\"{BASE_PATH}/gbif/processed/2024-05-01/occurrence_parquets\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "338e7ed8-6e53-4077-9520-2ad63b79bb7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# bash:\n",
    "# mongod --dbpath=\"${BASE_PATH}/TreeOfLife/gbif_mongo\" --fork --logpath=\"${BASE_PATH}/mongo_logs/gbif_mongo.log\"\n",
    "# Check process status\n",
    "# ps aux | grep mongod\n",
    "\n",
    "from pymongo import MongoClient\n",
    "\n",
    "# Step 1: Connect to the MongoDB server\n",
    "client = MongoClient(\"mongodb://localhost:27017/\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ade00f5-c730-42fe-a276-92f4b9aafa23",
   "metadata": {},
   "source": [
    "# Identify Camera-Trap Data within the Multi-Image Subset\n",
    "\n",
    "We use `basisOfRecord` and `eventRemarks` to identify camera-trap data entries.\n",
    "\n",
    "## Criteria for Camera-Trap Data\n",
    "\n",
    "1. Records with `eventRemarks` containing:\n",
    "  - `'camera trap'`\n",
    "  - `'Trigger Sensitivity'`\n",
    "Most of these records also have `basisOfRecord` labeled as `MACHINE_OBSERVATION`.\n",
    "    - **Exception**: 31 entries from Sibecocenter LLC are labeled as `HUMAN_OBSERVATION`.\n",
    "\n",
    "2. A list of publishers from `MACHINE_OBSERVATION` that don't have `eventRemarks`\n",
    "\n",
    "## `MACHINE_OBSERVATION` Data Breakdown\n",
    "**Number of `MACHINE_OBSERVATION` Records:**\n",
    "- Total: 235,632\n",
    "  - `eventRemarks` is null: 98,835\n",
    "  - `eventRemarks` contains `'camera trap'` or `'Trigger Sensitivity'`: 136,705\n",
    "  - `eventRemarks` contains weather-related information: 92\n",
    "\n",
    "**Note**:\n",
    "- Entries with weather-related `eventRemarks` are confirmed **not** to be camera-trap data.\n",
    "- Entries with `'camera trap'` or `'Trigger Sensitivity'` in `eventRemarks` are confirmed as camera-trap data.\n",
    "\n",
    "### Records with Null `eventRemarks`\n",
    "For records without `eventRemarks`, camera-trap data can be inferred from their images. I've reviewed them by publishers\n",
    "\n",
    "---\n",
    "\n",
    "#### Camera Trap Data Publishers\n",
    "- **National Museum of Nature and Science, Japan**\n",
    "  - *[Plankton image dataset](https://www.gbif.org/dataset/2b9b1484-431b-4dae-94bb-2259e6c1fdc6) from a cabled observatory system (JEDI System/OCEANS) deployed at coastal area of Oshima island, Tokyo, Japan*\n",
    "- **Museums Victoria**\n",
    "- **Ministerio del Ambiente, Agua y Transición Ecológica de Ecuador - MAATE**\n",
    "- **Miljøstyrelsen / The Danish Environmental Protection Agency**\n",
    "- **Burgoigee Creek Landcare Group**\n",
    "- **Lomonosov Moscow State University**\n",
    "\n",
    "---\n",
    "\n",
    "#### Non-Camera Trap Data Publishers\n",
    "- **Xeno-canto Foundation for Nature Sounds**\n",
    "  - *Audio files attached with images, not useful for training ([example](https://www.gbif.org/occurrence/4508370762))*\n",
    "- **Florida Museum of Natural History**\n",
    "  - *Citizen science, phone, archive ([example](https://www.gbif.org/occurrence/gallery?dataset_key=832b8188-f762-11e1-a439-00145eb45e9))*\n",
    "- **University of North Carolina at Chapel Hill Herbarium (NCU)**\n",
    "  - *Not camera trap data*\n",
    "- **Jardín Botánico de Quito**\n",
    "  - *Cell phone camera, not camera trap data*\n",
    "- **Negrita Films SAS**\n",
    "  - *High-res images, not camera trap data*\n",
    "- **University of Alaska Museum of the North**\n",
    "  - *Not camera trap data*\n",
    "- **University of Colorado Museum of Natural History**\n",
    "  - *High-res images, not camera trap data*\n",
    "\n",
    "---\n",
    "\n",
    "#### Unknown Publisher\n",
    "- **Área Metropolitana del Valle de Aburrá**: *Records could not be found.*\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3770f4b6-d2bd-4a92-92cf-b26bbc3fe590",
   "metadata": {},
   "source": [
    "Find subset of source ID that contains multiple images"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "a727ca4a-2a9f-436e-bc33-6f4ea6db3bff",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- eventRemarks: string (nullable = true)\n",
      " |-- publisher: string (nullable = true)\n",
      " |-- datasetKey: string (nullable = true)\n",
      " |-- basisOfRecord: string (nullable = true)\n",
      " |-- taxonomicStatus: string (nullable = true)\n",
      " |-- scientificName: string (nullable = true)\n",
      " |-- kingdom: string (nullable = true)\n",
      " |-- phylum: string (nullable = true)\n",
      " |-- class: string (nullable = true)\n",
      " |-- order: string (nullable = true)\n",
      " |-- family: string (nullable = true)\n",
      " |-- genus: string (nullable = true)\n",
      " |-- species: string (nullable = true)\n",
      " |-- distinct_uuid_count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "col_list = [\"source_id\", \"eventRemarks\", \"publisher\", \"datasetKey\", \"basisOfRecord\", \"taxonomicStatus\", \"scientificName\"] + COLS_TAXONOMIC\n",
    "source_id_with_multiple_uuids = (\n",
    "    spark_df.groupBy(*col_list)\n",
    "    .agg(\n",
    "        F.countDistinct(\"uuid\").alias(\"distinct_uuid_count\"),\n",
    "    )\n",
    "    .filter(col(\"distinct_uuid_count\") > 1)\n",
    ")\n",
    "source_id_with_multiple_uuids.cache()\n",
    "\n",
    "source_id_with_multiple_uuids.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "18eb3d69-e3bc-4dbc-b0a6-3e2315e71ea5",
   "metadata": {},
   "outputs": [],
   "source": [
    "cond_remarks = (\n",
    "    col(\"eventRemarks\").isNotNull() &\n",
    "    (\n",
    "        col(\"eventRemarks\").contains(\"camera trap\") |\n",
    "        col(\"eventRemarks\").contains(\"Trigger Sensitivity\")\n",
    "    )\n",
    ")\n",
    "publishers_camera_trap = [\n",
    "    #\"National Museum of Nature and Science, Japan\", # microscopic plankton images, excluded\n",
    "    \"Museums Victoria\",\n",
    "    \"Ministerio del Ambiente, Agua y Transición Ecológica de Ecuador - MAATE\",\n",
    "    \"Miljøstyrelsen / The Danish Environmental Protection Agency\",\n",
    "    \"Burgoigee Creek Landcare Group\",\n",
    "    \"Lomonosov Moscow State University\"\n",
    "]\n",
    "cond_publishers = (\n",
    "    col(\"basisOfRecord\") == \"MACHINE_OBSERVATION\"\n",
    ") & (\n",
    "    col(\"publisher\").isNotNull() & col(\"publisher\").isin(publishers_camera_trap)\n",
    ")\n",
    "\n",
    "cond_camera_trap = (cond_remarks | cond_publishers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "a3aea095-98b2-4195-81b5-b39a72c85266",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "assert (\n",
    "    source_id_with_multiple_uuids.filter(cond_camera_trap).count() + source_id_with_multiple_uuids.filter(~cond_camera_trap).count() == source_id_with_multiple_uuids.count()\n",
    "), \"The conditions do not fully partition the dataset.\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "e470c2e2-13da-41a8-9d94-c69c32ad4c82",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "assert (\n",
    "    spark_df.filter(cond_camera_trap).count() + spark_df.filter(~cond_camera_trap).count() == spark_df.count()\n",
    "), \"The conditions do not fully partition the dataset.\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "32dcb503-8753-4d50-933b-d0deb352a198",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "142266"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "source_id_with_multiple_uuids.filter(cond_camera_trap).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "445e193a-81a0-401e-8336-9c1227b5918f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+---------+\n",
      "|basisOfRecord      |count |bucket   |\n",
      "+-------------------+------+---------+\n",
      "|MACHINE_OBSERVATION|142235|100k-500k|\n",
      "|HUMAN_OBSERVATION  |31    |11-50    |\n",
      "+-------------------+------+---------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "view_freq(source_id_with_multiple_uuids.filter(cond_camera_trap), \"basisOfRecord\", truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9088a272-2210-4e92-8118-04e7488d499a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "(\n",
    "    spark_df\n",
    "    .join(\n",
    "        source_id_with_multiple_uuids.filter(cond_camera_trap).select(\"source_id\"),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    "    .select([\"uuid\", \"source_id\", \"datasetKey\", \"basisOfRecord\", \"publisher\"])\n",
    "    .repartition(4)\n",
    "    .write\n",
    "    .mode(\"overwrite\")\n",
    "    .parquet(f\"{BASE_PATH}/gbif/attributes/occurrence_camera_trap_multi_images\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "38228264-5195-4b72-b233-26bbe463cd23",
   "metadata": {},
   "source": [
    "# Invesitgate Dataset Metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9c9ef54-9676-471e-818f-a05cbadb32ef",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/attributes/occurrence_camera_trap_multi_images\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "796dbc2e-0302-42c5-a78a-9903aed5d4d7",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "data_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap/*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "c8cd16f8-d8fc-4fcb-9b64-c0f2e00dcfbb",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_camera_trap_filtered = data_camera_trap.filter(col(\"publisher\")!=\"National Museum of Nature and Science, Japan\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "984bfc69-8107-446f-9b38-b944be6c7f14",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- uuid: string (nullable = true)\n",
      " |-- original_size: array (nullable = true)\n",
      " |    |-- element: long (containsNull = true)\n",
      " |-- resized_size: array (nullable = true)\n",
      " |    |-- element: long (containsNull = true)\n",
      " |-- image: binary (nullable = true)\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- publisher: string (nullable = true)\n",
      " |-- kingdom: string (nullable = true)\n",
      " |-- phylum: string (nullable = true)\n",
      " |-- class: string (nullable = true)\n",
      " |-- order: string (nullable = true)\n",
      " |-- family: string (nullable = true)\n",
      " |-- genus: string (nullable = true)\n",
      " |-- species: string (nullable = true)\n",
      " |-- path: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data_camera_trap.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "9097cba4-958e-4c0e-8292-4462b54027db",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- uuid: string (nullable = true)\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- datasetKey: string (nullable = true)\n",
      " |-- basisOfRecord: string (nullable = true)\n",
      " |-- publisher: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_camera_trap.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "024352bd-b519-41f4-83f5-fbd68d4697b6",
   "metadata": {},
   "source": [
    "Total images count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "50755102-555e-4beb-8660-8ff2b5eb3dc6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "2433739"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_camera_trap.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a19872d1-f53a-4d18-86d0-82236a34decd",
   "metadata": {},
   "source": [
    "Total occurrence count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "c0c02f09-a8a6-4ce0-9469-baff37de2d43",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "208899"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_camera_trap.select(\"source_id\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "262318c7-b449-4fc7-a86a-83eee6ebb7a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_df = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        df_camera_trap,\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "016f33a2-bc64-42e3-9069-784ef886bf80",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate boolean columns and cast to integers for aggregation\n",
    "spark_df_with_flags = filtered_df.withColumn(\n",
    "    \"is_taxon_higher_rank\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_HIGHERRANK\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_taxon_match_none\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_NONE\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_taxon_match_fuzzy\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_FUZZY\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_any\", \n",
    "    when((col(\"issue\").contains(\"TAXON_MATCH_FUZZY\")) | (col(\"issue\").contains(\"TAXON_MATCH_NONE\")) | (col(\"issue\").contains(\"TAXON_MATCH_HIGHERRANK\")), 1).otherwise(0)\n",
    ")\n",
    "\n",
    "# Calculate sum and mean for each numeric column\n",
    "result_df = spark_df_with_flags.agg(\n",
    "    spark_sum(\"is_taxon_higher_rank\").alias(\"sum_taxon_higher_rank\"),\n",
    "    mean(\"is_taxon_higher_rank\").alias(\"mean_taxon_higher_rank\"),\n",
    "    spark_sum(\"is_taxon_match_none\").alias(\"sum_taxon_match_none\"),\n",
    "    mean(\"is_taxon_match_none\").alias(\"mean_taxon_match_none\"),\n",
    "    spark_sum(\"is_taxon_match_fuzzy\").alias(\"sum_taxon_match_fuzzy\"),\n",
    "    mean(\"is_taxon_match_fuzzy\").alias(\"mean_taxon_match_fuzzy\"),\n",
    "    spark_sum(\"is_any\").alias(\"sum_is_any\"),\n",
    "    mean(\"is_any\").alias(\"mean_is_any\")\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "f7f48d0e-96ef-4c14-94e4-9e724aae9a10",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 74:===================================================>  (191 + 9) / 200]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+-------------------+\n",
      "|sum_taxon_higher_rank|mean_taxon_higher_rank|sum_taxon_match_none|mean_taxon_match_none|sum_taxon_match_fuzzy|mean_taxon_match_fuzzy|sum_is_any|mean_is_any        |\n",
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+-------------------+\n",
      "|8692                 |0.0035714593882088426 |69279               |0.028466076271942062 |355                  |1.4586609328280477E-4 |78326     |0.03218340175343371|\n",
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+-------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "result_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05364788-b903-470e-b776-77d1c71c9b76",
   "metadata": {},
   "source": [
    "| Category               | Count  | Percentage  |\n",
    "|------------------------|--------|-------------|\n",
    "| Taxon Higher Rank      | 8,692  | 0.36%       |\n",
    "| Taxon Match None       | 69,279 | 2.85%       |\n",
    "| Taxon Match Fuzzy      | 355    | 0.01%       |\n",
    "| Contains Any           | 78,326 | 3.22%       |\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "762ed6fa-381c-4a27-a459-a1987543130e",
   "metadata": {},
   "source": [
    "# Distribution Analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90bdc124-d131-4549-baa9-8039c3f75142",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "lookup_tbl_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_multi_images_camera_trap\")\n",
    "lookup_tbl = spark.read.parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_tables\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "58a65a0a-da15-41a2-b465-4ce229fe70ca",
   "metadata": {},
   "outputs": [],
   "source": [
    "lookup_tbl_camera_trap = lookup_tbl_camera_trap.filter(col(\"publisher\") != \"National Museum of Nature and Science, Japan\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "b726f7ea-ce30-4ff4-9f3f-a67c274c46a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "COLS_TAXONOMIC_KEY = [x + \"Key\" for x in COLS_TAXONOMIC]\n",
    "\n",
    "lookup_tbl_camera_trap = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        spark_df.select([\"uuid\"] + COLS_TAXONOMIC_KEY),\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")\n",
    "\n",
    "spark_df = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        lookup_tbl.select(\"uuid\"),\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "f5f57ec2-92ee-4e87-b5c1-570c90b7bec0",
   "metadata": {},
   "outputs": [],
   "source": [
    "agg_df = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .groupBy([\"source_id\"] + COLS_TAXONOMIC_KEY)\n",
    "    .count()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "602c07c0-ff3d-457e-b6fe-8a0cbaf454aa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- kingdomKey: string (nullable = true)\n",
      " |-- phylumKey: string (nullable = true)\n",
      " |-- classKey: string (nullable = true)\n",
      " |-- orderKey: string (nullable = true)\n",
      " |-- familyKey: string (nullable = true)\n",
      " |-- genusKey: string (nullable = true)\n",
      " |-- speciesKey: string (nullable = true)\n",
      " |-- count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "agg_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "60a61c6f-2158-4d0a-a646-b453e70a621c",
   "metadata": {},
   "outputs": [],
   "source": [
    "threshold = 15\n",
    "\n",
    "occ_keep_df = agg_df.filter(col(\"count\")<=threshold)\n",
    "occ_omit_df = agg_df.filter(col(\"count\")>threshold)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "edee9b30-c7c4-4ad6-bb57-a66ffa3719dc",
   "metadata": {},
   "outputs": [],
   "source": [
    "taxon_level = \"species\"\n",
    "taxon_level_key = taxon_level + \"Key\"\n",
    "\n",
    "taxon_freq = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        occ_omit_df.select([\"source_id\"]),\n",
    "        on = \"source_id\",\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "taxon_freq_camera_trap_keep = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        occ_omit_df.select([\"source_id\"]),\n",
    "        on = \"source_id\",\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "taxon_freq_camera_trap_omit = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        occ_omit_df.select([\"source_id\"]),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "f55042c7-19ab-4e42-8567-c5e57a3e4413",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- species: string (nullable = true)\n",
      " |-- speciesKey: string (nullable = true)\n",
      " |-- img_count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "taxon_freq_camera_trap_omit.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "42a1fa6d-e065-42e9-acfb-f5ffc95c9baa",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 38:====================================================> (194 + 6) / 200]0]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Removed occ: 26675\n",
      "Removed occ with no species label: 4530\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "lookup_tbl_camera_trap_omit = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        occ_omit_df.select([\"source_id\"]),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")\n",
    "\n",
    "n_omit_occ = lookup_tbl_camera_trap_omit.select([\"source_id\"]).distinct().count()\n",
    "n_omit_occ_species_null = lookup_tbl_camera_trap_omit.filter(col(\"species\").isNull()).select([\"source_id\"]).distinct().count()\n",
    "\n",
    "print(f\"Removed occ: {n_omit_occ}\")\n",
    "print(f\"Removed occ with no species label: {n_omit_occ_species_null}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "5cdf5268-e1ac-4939-a442-511772853a02",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- kingdomKey: string (nullable = true)\n",
      " |-- phylumKey: string (nullable = true)\n",
      " |-- classKey: string (nullable = true)\n",
      " |-- orderKey: string (nullable = true)\n",
      " |-- familyKey: string (nullable = true)\n",
      " |-- genusKey: string (nullable = true)\n",
      " |-- speciesKey: string (nullable = true)\n",
      " |-- count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "occ_omit_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "0faa587c-2743-427e-af64-6feaa7ecb51c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+\n",
      "|column_name|sparsity            |\n",
      "+-----------+--------------------+\n",
      "|kingdomKey |0.0                 |\n",
      "|phylumKey  |0.0                 |\n",
      "|classKey   |6.621054954756124E-4|\n",
      "|orderKey   |4.414036636504083E-4|\n",
      "|familyKey  |0.16883690134628118 |\n",
      "|genusKey   |0.1703818141690576  |\n",
      "|speciesKey |1.0                 |\n",
      "+-----------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "check_sparsity(occ_omit_df.filter(col(\"speciesKey\").isNull()).select(COLS_TAXONOMIC_KEY)).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c245819-9c8b-4a33-87b8-463226993ac8",
   "metadata": {},
   "source": [
    "If species label is missing, roll-up a level to genus"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "5c733de8-4e17-45aa-9a29-b26236394745",
   "metadata": {},
   "outputs": [],
   "source": [
    "taxon_level = \"genusKey\"\n",
    "taxon_level_key = \"genusKey\"\n",
    "\n",
    "taxon_freq = (\n",
    "    spark_df\n",
    "    .join(occ_omit_df.filter((col(\"speciesKey\").isNull())&(col(\"genusKey\").isNotNull())).select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "# Camera trap - Kept occurrences\n",
    "taxon_freq_camera_trap_keep = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(occ_omit_df.filter((col(\"speciesKey\").isNull())&(col(\"genusKey\").isNotNull())).select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "# Camera trap - Omitted occurrences\n",
    "taxon_freq_camera_trap_omit = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(occ_omit_df.filter((col(\"speciesKey\").isNull())&(col(\"genusKey\").isNotNull())).select([\"source_id\"]), on=\"source_id\", how=\"inner\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "a9e9caca-b02b-41ba-9b71-74006dbcb7b1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 380:===================================================> (195 + 5) / 200]0]0]]]]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Genus loss in the camera-trap subset: 2\n",
      "Genus loss in the GBIF OCC: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "taxon_loss_camera_trap_kept = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(taxon_freq_camera_trap_keep, on=taxon_level_key, how=\"left_anti\")\n",
    ").count()\n",
    "\n",
    "taxon_loss_gbif = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(taxon_freq, on=taxon_level_key, how=\"left_anti\")\n",
    ").count()\n",
    "\n",
    "print(f\"Genus loss in the camera-trap subset: {taxon_loss_camera_trap_kept}\")\n",
    "print(f\"Genus loss in the GBIF OCC: {taxon_loss_gbif}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3b61c5f3-b3b4-4568-a55a-e3bcf7950c4c",
   "metadata": {},
   "source": [
    "With the discarded subset:\n",
    "- How many unique `taxon_level` can't found in the kept subset\n",
    "- How many unique `taxon_level` can't found in the GBIF occurrence (successfully downloaded)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "c33a87cb-cded-4326-8a5d-051822ec5f59",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                0]]\r"
     ]
    }
   ],
   "source": [
    "n_camera_trap_kept = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(\n",
    "        taxon_freq_camera_trap_keep,\n",
    "        on = \"speciesKey\",\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    ").count()\n",
    "\n",
    "n_total = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(\n",
    "        taxon_freq,\n",
    "        on = \"speciesKey\",\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    ").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1a4a4f5-d39f-499a-bbb3-c9d2eecde9ec",
   "metadata": {},
   "source": [
    "Run simulations by changing the threshold"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "d6383e51-10ef-421d-81b1-e12e7948d351",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- kingdomKey: string (nullable = true)\n",
      " |-- phylumKey: string (nullable = true)\n",
      " |-- classKey: string (nullable = true)\n",
      " |-- orderKey: string (nullable = true)\n",
      " |-- familyKey: string (nullable = true)\n",
      " |-- genusKey: string (nullable = true)\n",
      " |-- speciesKey: string (nullable = true)\n",
      " |-- count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "occ_omit_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c8ed010-ea29-4600-af4d-585583b7bfec",
   "metadata": {},
   "outputs": [],
   "source": [
    "lookup_tbl_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_multi_images_camera_trap\")\n",
    "lookup_tbl = spark.read.parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_tables\")\n",
    "\n",
    "COLS_TAXONOMIC_KEY = [x + \"Key\" for x in COLS_TAXONOMIC]\n",
    "\n",
    "lookup_tbl_camera_trap = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        spark_df.select([\"uuid\"] + COLS_TAXONOMIC_KEY),\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")\n",
    "\n",
    "spark_df = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        lookup_tbl.select(\"uuid\"),\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")\n",
    "\n",
    "def run_simulation(spark_df, lookup_tbl_camera_trap, threshold_values, taxon_level=\"species\"):\n",
    "    \"\"\"\n",
    "    Runs a simulation to assess taxon loss at different occurrence count thresholds.\n",
    "    \n",
    "    Parameters:\n",
    "        spark_df (DataFrame): Original occurrence data with taxon information.\n",
    "        lookup_tbl_camera_trap (DataFrame): Camera trap lookup table.\n",
    "        taxon_level (str): Taxonomic level to analyze (default is \"species\").\n",
    "    \n",
    "    Returns:\n",
    "        DataFrame: Spark DataFrame containing simulation results.\n",
    "    \"\"\"\n",
    "    #threshold_values = list(range(60, 10, -10)) + [15, 12, 10, 8, 5, 2]\n",
    "    taxon_level_key = taxon_level + \"Key\"\n",
    "    simulation_results = []\n",
    "\n",
    "    agg_df = (\n",
    "        lookup_tbl_camera_trap\n",
    "        .groupBy([\"source_id\"] + COLS_TAXONOMIC_KEY)\n",
    "        .count()\n",
    "    )\n",
    "        \n",
    "    for threshold in tqdm(threshold_values, desc=\"Running Simulations\"):\n",
    "        occ_keep_df = agg_df.filter(col(\"count\") <= threshold)\n",
    "        occ_omit_df = agg_df.filter(col(\"count\") > threshold)\n",
    "\n",
    "        n_occ_omit = occ_omit_df.select(\"source_id\").distinct().count()\n",
    "        n_img_omit = (\n",
    "            lookup_tbl_camera_trap\n",
    "            .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"inner\")\n",
    "            .select(\"uuid\").distinct().count()\n",
    "        )\n",
    "        \n",
    "        # Calculate taxon frequencies for kept occurrences\n",
    "        taxon_freq = (\n",
    "            spark_df\n",
    "            .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "            .select([taxon_level, taxon_level_key])\n",
    "            .groupBy([taxon_level, taxon_level_key])\n",
    "            .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "            .filter(col(taxon_level).isNotNull())\n",
    "        )\n",
    "\n",
    "        # Camera trap - Kept occurrences\n",
    "        taxon_freq_camera_trap_keep = (\n",
    "            lookup_tbl_camera_trap\n",
    "            .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "            .select([taxon_level, taxon_level_key])\n",
    "            .groupBy([taxon_level, taxon_level_key])\n",
    "            .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "            .filter(col(taxon_level).isNotNull())\n",
    "        )\n",
    "\n",
    "        # Camera trap - Omitted occurrences\n",
    "        taxon_freq_camera_trap_omit = (\n",
    "            lookup_tbl_camera_trap\n",
    "            .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"inner\")\n",
    "            .select([taxon_level, taxon_level_key])\n",
    "            .groupBy([taxon_level, taxon_level_key])\n",
    "            .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "            .filter(col(taxon_level).isNotNull())\n",
    "        )\n",
    "\n",
    "        taxon_loss_camera_trap_kept = (\n",
    "            taxon_freq_camera_trap_omit\n",
    "            .join(taxon_freq_camera_trap_keep, on=taxon_level_key, how=\"left_anti\")\n",
    "        ).count()\n",
    "\n",
    "        taxon_loss_gbif = (\n",
    "            taxon_freq_camera_trap_omit\n",
    "            .join(taxon_freq, on=taxon_level_key, how=\"left_anti\")\n",
    "        ).count()\n",
    "\n",
    "        # Store results\n",
    "        simulation_results.append((\n",
    "            threshold, taxon_loss_camera_trap_kept, taxon_loss_gbif,\n",
    "            n_occ_omit, n_img_omit\n",
    "        ))\n",
    "    \n",
    "    simulation_df = spark.createDataFrame(\n",
    "        simulation_results, \n",
    "        [\"threshold\", \"taxon_loss_camera_trap_kept\", \"taxon_loss_gbif\", \"n_occ_omit\", \"n_img_omit\"]\n",
    "    )\n",
    "    \n",
    "    return simulation_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "afbfdfed-50ef-40de-ba0f-e65e4b7623e3",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Running Simulations: 100%|██████████| 11/11 [14:25<00:00, 78.66s/it]             200]]0]]\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_df = run_simulation(\n",
    "    spark_df, lookup_tbl_camera_trap,\n",
    "    threshold_values = list(range(60, 10, -10)) + [15, 12, 10, 8, 5, 2],\n",
    "    taxon_level=\"species\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "3c51629d-cdd4-46f0-b4fc-d7010e1d9a13",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|threshold|taxon_loss_camera_trap_kept|taxon_loss_gbif|n_occ_omit|n_img_omit|\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|60       |2                          |0              |2091      |359070    |\n",
      "|50       |4                          |0              |2853      |403854    |\n",
      "|40       |7                          |0              |4172      |468362    |\n",
      "|30       |12                         |0              |6682      |565931    |\n",
      "|20       |23                         |0              |12393     |729985    |\n",
      "|15       |41                         |0              |26676     |1012534   |\n",
      "|12       |59                         |1              |28220     |1034344   |\n",
      "|10       |78                         |1              |29796     |1052456   |\n",
      "|8        |137                        |2              |115417    |1903704   |\n",
      "|5        |212                        |6              |124541    |1961861   |\n",
      "|2        |534                        |16             |141493    |2023422   |\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f60d1cbe-5a8a-4b42-b758-ff01490c8f39",
   "metadata": {},
   "source": [
    "TODO: Are they `taxon_loss_camera_trap_kept` in lila?\n",
    "\n",
    "Check species NULL, set threshold to 15+ order"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "e5d033ef-4adb-4712-8b6e-c8f0efac8bcb",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Running Simulations: 100%|██████████| 11/11 [13:38<00:00, 74.38s/it]             200]]50]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.25 s, sys: 319 ms, total: 1.56 s\n",
      "Wall time: 13min 38s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_df_genus = run_simulation(\n",
    "    spark_df, lookup_tbl_camera_trap,\n",
    "    threshold_values = list(range(60, 10, -10)) + [15, 12, 10, 8, 5, 2],\n",
    "    taxon_level=\"genus\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "e74294d5-2c0c-4ae3-8275-4595af0a567e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|threshold|taxon_loss_camera_trap_kept|taxon_loss_gbif|n_occ_omit|n_img_omit|\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|60       |1                          |0              |2091      |359070    |\n",
      "|50       |2                          |0              |2853      |403854    |\n",
      "|40       |3                          |0              |4172      |468362    |\n",
      "|30       |8                          |0              |6682      |565931    |\n",
      "|20       |15                         |0              |12393     |729985    |\n",
      "|15       |26                         |0              |26676     |1012534   |\n",
      "|12       |34                         |0              |28220     |1034344   |\n",
      "|10       |47                         |0              |29796     |1052456   |\n",
      "|8        |90                         |0              |115417    |1903704   |\n",
      "|5        |162                        |0              |124541    |1961861   |\n",
      "|2        |445                        |5              |141493    |2023422   |\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_df_genus.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "409a09e7-2227-4e36-9891-e149f7678834",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Running Simulations: 100%|██████████| 11/11 [13:39<00:00, 74.51s/it]             / 450]0]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.18 s, sys: 336 ms, total: 1.51 s\n",
      "Wall time: 13min 39s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_df_family = run_simulation(\n",
    "    spark_df, lookup_tbl_camera_trap,\n",
    "    threshold_values = list(range(60, 10, -10)) + [15, 12, 10, 8, 5, 2],\n",
    "    taxon_level=\"family\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "694396a3-abd9-4f6b-b472-c15200c11fdf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|threshold|taxon_loss_camera_trap_kept|taxon_loss_gbif|n_occ_omit|n_img_omit|\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|60       |1                          |0              |2091      |359070    |\n",
      "|50       |1                          |0              |2853      |403854    |\n",
      "|40       |1                          |0              |4172      |468362    |\n",
      "|30       |3                          |0              |6682      |565931    |\n",
      "|20       |6                          |0              |12393     |729985    |\n",
      "|15       |11                         |0              |26676     |1012534   |\n",
      "|12       |13                         |0              |28220     |1034344   |\n",
      "|10       |18                         |0              |29796     |1052456   |\n",
      "|8        |33                         |0              |115417    |1903704   |\n",
      "|5        |56                         |0              |124541    |1961861   |\n",
      "|2        |186                        |1              |141493    |2023422   |\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_df_family.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "fda3d7fd-3ada-49a3-bd27-bcaa16e86664",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Running Simulations: 100%|██████████| 11/11 [13:38<00:00, 74.39s/it]             / 450]0]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.09 s, sys: 339 ms, total: 1.43 s\n",
      "Wall time: 13min 38s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_df_order = run_simulation(\n",
    "    spark_df, lookup_tbl_camera_trap,\n",
    "    threshold_values = list(range(60, 10, -10)) + [15, 12, 10, 8, 5, 2],\n",
    "    taxon_level=\"order\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "5eaa6cec-a930-4294-9f5a-5dbd847d588b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|threshold|taxon_loss_camera_trap_kept|taxon_loss_gbif|n_occ_omit|n_img_omit|\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "|60       |0                          |0              |2091      |359070    |\n",
      "|50       |0                          |0              |2853      |403854    |\n",
      "|40       |0                          |0              |4172      |468362    |\n",
      "|30       |2                          |0              |6682      |565931    |\n",
      "|20       |3                          |0              |12393     |729985    |\n",
      "|15       |5                          |0              |26676     |1012534   |\n",
      "|12       |7                          |0              |28220     |1034344   |\n",
      "|10       |8                          |0              |29796     |1052456   |\n",
      "|8        |11                         |0              |115417    |1903704   |\n",
      "|5        |16                         |0              |124541    |1961861   |\n",
      "|2        |48                         |0              |141493    |2023422   |\n",
      "+---------+---------------------------+---------------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_df_order.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "421b7b7d-3e03-4a74-ba25-2c4532131217",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                0]\r"
     ]
    }
   ],
   "source": [
    "occ_keep_df = agg_df.filter(col(\"count\") <= 15)\n",
    "occ_omit_df = agg_df.filter(col(\"count\") > 15)\n",
    "\n",
    "n_occ_omit = occ_omit_df.select(\"source_id\").distinct().count()\n",
    "n_img_omit = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"inner\")\n",
    "    .select(\"uuid\").distinct().count()\n",
    ")\n",
    "\n",
    "# Calculate taxon frequencies for kept occurrences\n",
    "taxon_freq = (\n",
    "    spark_df\n",
    "    .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "# Camera trap - Kept occurrences\n",
    "taxon_freq_camera_trap_keep = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"left_anti\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "# Camera trap - Omitted occurrences\n",
    "taxon_freq_camera_trap_omit = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(occ_omit_df.select([\"source_id\"]), on=\"source_id\", how=\"inner\")\n",
    "    .select([taxon_level, taxon_level_key])\n",
    "    .groupBy([taxon_level, taxon_level_key])\n",
    "    .count().withColumnRenamed(\"count\", \"img_count\")\n",
    "    .filter(col(taxon_level).isNotNull())\n",
    ")\n",
    "\n",
    "taxon_loss_camera_trap_kept = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(taxon_freq_camera_trap_keep, on=taxon_level_key, how=\"left_anti\")\n",
    ")\n",
    "\n",
    "taxon_loss_gbif = (\n",
    "    taxon_freq_camera_trap_omit\n",
    "    .join(taxon_freq, on=taxon_level_key, how=\"left_anti\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "e453742c-7af9-4642-b3b3-0465d7235185",
   "metadata": {},
   "outputs": [],
   "source": [
    "occ_omit_species_loss_df = (\n",
    "    occ_omit_df\n",
    "    .join(\n",
    "        taxon_loss_camera_trap_kept.select(\"speciesKey\"),\n",
    "        on = \"speciesKey\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    "    .select(\"source_id\")\n",
    "    .join(\n",
    "        lookup_tbl_camera_trap.select([\"source_id\"] + COLS_TAXONOMIC).distinct(),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "3c515e60-e6e4-4e8c-9872-713dc8da66c6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[source_id: string, kingdom: string, phylum: string, class: string, order: string, family: string, genus: string, species: string]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "occ_omit_species_loss_df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "756a748b-dc01-4232-9409-dd2ca5b24aa1",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "de0b65b1-d745-48a5-b951-0e5514e0eee9",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_subset = (\n",
    "    df_camera_trap\n",
    "    .join(\n",
    "        occ_omit_species_loss_df.select(\"source_id\"),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a7c6a243-4548-4b64-b010-a1f8923a985d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_subset.write.parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap_below_15_species_loss\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8336bd4d-5e60-4072-acd4-bae5f426d53c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_subset = spark.read.parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap_below_15_species_loss\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b4f230a4-0cfd-4330-8c6e-588d5ced9f12",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_subset.drop(\"image\", \"original_size\", \"resized_size\").toPandas().to_csv(\n",
    "    f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap_below_15_species_loss.csv\",\n",
    "    index=False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d9b19d7b-4a44-41b2-8c71-a3d31d34a0f0",
   "metadata": {},
   "outputs": [],
   "source": [
    "metadata_all = spark.read.parquet(f\"{BASE_PATH}/TreeOfLife/metadata\")\n",
    "metadata_lila = metadata_all.filter(col(\"source\").isin([\"lila-bc-multi-label\", \"lila-bc-single-label\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "b15d87aa-8341-4387-8551-2cc7a2563306",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(\n",
    "    df_subset\n",
    "    .select(\"species\").distinct()\n",
    "    .join(\n",
    "        metadata_lila.select(\"species\").distinct(),\n",
    "        on = \"species\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3c2df0b0-063d-4532-b86c-715ec241f1bb",
   "metadata": {},
   "source": [
    "# Filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "3ae9cd97-2bdf-42c0-8195-f16562ef31b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "threshold = 15\n",
    "\n",
    "occ_keep_df = agg_df.filter(col(\"count\")<=threshold)\n",
    "occ_omit_df = agg_df.filter(col(\"count\")>threshold)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "0d04dd8f-2484-4032-9dae-f661aec9588e",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "2056359"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lookup_tbl_camera_trap.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "7757e9c6-deaf-4b62-a541-a3f77e139c0e",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                0]\r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "1010242"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lookup_tbl_camera_trap_15 = (\n",
    "    lookup_tbl_camera_trap\n",
    "    .join(\n",
    "        occ_keep_df.select(\"source_id\"),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")\n",
    "lookup_tbl_camera_trap_15.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e89657b5-38fc-48b1-8786-65254feb989f",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_camera_trap = spark.read.parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "e3ce0cef-ba95-4f2f-bba3-4ead75400d89",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_camera_trap_15 = (\n",
    "    df_camera_trap\n",
    "    .join(\n",
    "        occ_keep_df.select(\"source_id\"),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ce84a3f-4b70-4c57-8071-2dd84d6a3913",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                450]\r"
     ]
    }
   ],
   "source": [
    "df_camera_trap_15.write.mode(\"overwrite\").parquet(f\"{BASE_PATH}/gbif/image_lookup/multi_images_camera_trap_below_15\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "14bdf3ee-d10b-4db5-8a8d-4f84414eabc8",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "jupytext": {
   "formats": "ipynb,py:percent"
  },
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pyspark"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
